package com.jiuzhou.crawlers.taobao.util;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class HdfsUtil {

	static {
		String hadoopHome = System.getenv("HADOOP_HOME");
		if (hadoopHome != null && !hadoopHome.isEmpty())
			System.setProperty("hadoop.home.dir", hadoopHome);
	}
	// 在指定位置新建文件并写入字符
	public static synchronized void WriteToHdfs(String filePathName, String content) {
		FileSystem fs = null;
		FSDataOutputStream out = null;
		try {
			fs = FileSystem.get(URI.create(filePathName), new Configuration());
			out = fs.create(new Path(filePathName));
			out.write(content.getBytes("UTF-8"));
			// or using ----out.writeBytes(content);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (out != null) {
				try {
					out.close();
					out = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (fs != null) {
				try {
					fs.close();
					fs = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	// 从一个流里面写入
	// 如果是要从输入流中写入，或是从一个文件写到另一个文件（此时用输入流打开已有内容的文件）
	// 可以使用如下IOUtils.copyBytes方法。
	public static synchronized void WriteToHdfsFromStream(String fromFile, String toFile) {
		FileSystem fs = null;
		FSDataOutputStream out = null;
		try {
			fs = FileSystem.get(URI.create(toFile), new Configuration());
			out = fs.create(new Path(toFile));
			Path path = new Path(fromFile);
			FSDataInputStream in = fs.open(path);
			IOUtils.copyBytes(in, out, fs.getFileStatus(path).getLen(), true); // true表示复制完成后关闭流

		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (out != null) {
				try {
					out.close();
					out = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (fs != null) {
				try {
					fs.close();
					fs = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static synchronized String ReadFromHdfs(String filePathName) {
		FileSystem fs = null;
		FSDataInputStream in = null;
		try{
			fs = FileSystem.get(URI.create(filePathName), new Configuration());
			Path path = new Path(filePathName);
			in = fs.open(path);
			// IOUtils.copyBytes(in, System.out,
			// fs.getFileStatus(path).getLen(),true);
			byte[] b = new byte[(int) fs.getFileStatus(path).getLen()];
			in.readFully(b);
			return new String(b);
		} catch (IOException e) {
			e.printStackTrace();
			return "";
		} finally {
			if (in != null) {
				try {
					in.close();
					in = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (fs != null) {
				try {
					fs.close();
					fs = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 本地文件上传
	 */
	public static synchronized void uploadToHdfs(String localFileName, String remoteFileName) {
		FileSystem fs = null;
		try {
			fs = FileSystem.get(URI.create(remoteFileName), new Configuration());
			fs.copyFromLocalFile(new Path(localFileName), new Path(remoteFileName));
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (fs != null) {
				try {
					fs.close();
					fs = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 本地文件下载
	 */
	public static synchronized void downloadToHdfs(String remoteFileName, String localFileName) {
		FileSystem fs = null;
		try {
			fs = FileSystem.get(URI.create(remoteFileName), new Configuration());
			fs.copyToLocalFile(false, new Path(remoteFileName), new Path(localFileName), true);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (fs != null) {
				try {
					fs.close();
					fs = null;
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	 /**方法名称：getFilesUnderFolder()
	 * 方法说明：获取路径下所有的文件名称
	 * @param folderPath
	 * @param pattern
	 * @return 文件名
	 * @throws IOException List<Path>
	 */
	public  synchronized static List<Path> getFilesUnderFolder(Path folderPath, String pattern) throws IOException {  
		FileSystem fs = null;
		fs = FileSystem.get(URI.create(folderPath.toString()), new Configuration());
		  List<Path> paths = new ArrayList<Path>();  
		  if (fs.exists(folderPath)) {  
		   FileStatus[] fileStatus = fs.listStatus(folderPath);  
		   for (int i = 0; i < fileStatus.length; i++) {  
		    FileStatus fileStatu = fileStatus[i];  
		   // if (!fileStatu.isDir()) {//只要文件  
		     Path oneFilePath = fileStatu.getPath();  
		     if (pattern == null && !oneFilePath.getName().contains("detail")) {  
		      paths.add(oneFilePath);  
		     } else {  
		      if (oneFilePath.getName().contains(pattern) && !oneFilePath.getName().contains("detail")) {  
		       paths.add(oneFilePath);  
		      }  
		     }    
		  
		  //  }  
		   }  
		  }  
		  return paths;  
		 }  

	 /**方法名称：readByLineParamName()
	 * 方法说明：根据行变量名获取文件名称
	 * @param filePath
	 * @param paramName
	 * @return 文件名称
	 * @throws IOException List<String>
	 */
	public synchronized static List<String> readByLineParamName(Path filePath,String paramName) throws IOException{
			FileSystem fs = null;
			FSDataInputStream in = null;
			BufferedReader br =null;  
			List<String> allIdLines =new ArrayList<String>();
			String line = "";
				fs = FileSystem.get(URI.create(filePath.toString()), new Configuration());
				in = fs.open(filePath);
			    br = new BufferedReader(new InputStreamReader(in));  
			       while ((line = br.readLine()) != null) {  
			    	  if(line.contains(paramName)){
			    		  allIdLines.add(line);
			    	  }//end of if
            }//end of while
				return allIdLines;
	 }//end of method
	 
	
	public synchronized static List<List<Path>> getAllFileByRootPath(String rootPath){
		Path toPath = new Path(rootPath);//转化为Path类型
		List<List<Path>> dataFilePaths = new  ArrayList<List<Path>>(); //用list集合存放所有爬取类型的
			 try {
				List<Path> typeNamePaths = HdfsUtil.getFilesUnderFolder(toPath, "");

				for(Path root : typeNamePaths){
					dataFilePaths.add(HdfsUtil.getFilesUnderFolder(root, ""));
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		return dataFilePaths;
		
	}
}//end of class
